package com.atguigu.edu.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.EduConfig;
import com.atguigu.edu.realtime.util.DruidDSUtil;
import com.atguigu.edu.realtime.util.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;

/**
 * @ClassName DimSinkFunction
 * @Description TODO
 * @Author AFei
 * @Date 2022/9/4 16:49
 * @Version 1.0
 */
public class DimSinkFunction extends RichSinkFunction<com.alibaba.fastjson.JSONObject> {
    private DruidDataSource dataSource;

    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidDSUtil.createDataSource();
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        // 获取目的地表名
        String sinkTable = jsonObj.getString("sink_table");
        jsonObj.remove("sink_table");
        String type = jsonObj.getString("type");
        jsonObj.remove("type");
        // 拼接upsert语句
        String upsertSql = "upsert into "+ EduConfig.PHOENIX_SCHEMA +"."+sinkTable +
                " ("+ StringUtils.join(jsonObj.keySet(),",")+") " +
                " values" +
                " ('"+StringUtils.join(jsonObj.values(),"','")+"')";
        System.out.println("向phoenix表中插入数据的SQL：" + upsertSql);

        Connection conn = dataSource.getConnection();
        PhoenixUtil.executeSql(conn,upsertSql);

    }
}
